
#include "zeppreactor.h"
#include <stdio.h>

int accept_cb(int fd, uint32_t events, void *arg) {
    if (arg == NULL)
        return -1;

    struct zeppreactor *reactor = (struct zeppreactor *)arg;

    struct sockaddr_in clientaddr;
    socklen_t len = sizeof(clientaddr);

    int clientfd = accept(fd, (struct sockaddr *)&clientaddr, &len);
    if (clientfd < 0) {
        if (errno != EAGAIN && errno != EINTR) {
            printf("accept: %s\n", strerror(errno));
            return -1;
        }
    }

#if C100K
    int i;
    struct zeppevent *ev = zeppreactor_idx(reactor, clientfd);

    int flag = fcntl(clientfd, F_SETFL, O_NONBLOCK);  // 将clientfd设置为非阻塞
    if (flag < 0) {
        printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
        return -1;
    }
    zeppevent_set(ev, clientfd, recv_cb, reactor);
    zeppevent_add(reactor->epfd, EPOLLIN, ev);

#if BENCHMARKING
    if (clientfd % 1000 == 999) {
        struct timeval tv_cur;

        gettimeofday(&tv_begin, NULL);
        int time_used = TIME_SUB_MS(tv_begin, tv_cur);

        memcpy(&tv_cur, &tv_begin, sizeof(struct timeval));

        printf("--->>>clientfd: %d, time_used: %d\n", clientfd, time_used);
    }
#else
    printf("new connect [%s:%d][time:%ld], pos[%d]\n",
           inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port), ev->last_active, clientfd);
#endif

#else
    int i;
    do {
        for (i = 3; i < MAX_EPOLL_EVENTS; i++) {
            if (reactor->events[i].status == 0) {
                break;
            }
        }
        if (i == MAX_EPOLL_EVENTS) {
            printf("%s: max connect limit[%d]\n", __func__, MAX_EPOLL_EVENTS);
            break;
        }
        int flag = fcntl(clientfd, F_SETFL, O_NONBLOCK);  // 将clientfd设置为非阻塞
        if (flag < 0) {
            printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
            break;
        }

        zeppevent_set(&reactor->events[clientfd], clientfd, recv_cb, reactor);
        zeppevent_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]);
    } while (0);

    printf("new connect [%s:%d][time:%ld], pos[%d]\n",
           inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port), reactor->events[clientfd].last_active, clientfd);
#endif
    return 1;
}


int recv_cb(int fd, uint32_t events, void *arg) {
    if (arg == NULL)
        return -1;

    struct zeppreactor *reactor = (struct zeppreactor *)arg;
#if C100K
    struct zeppevent *ev = zeppreactor_idx(reactor, fd);
#else
    struct zeppevent *ev = reactor->events + fd;
#endif

    ssize_t len = recv(fd, ev->buffer, BUFFER_LENGTH, 0);

    // del --> ?
    zeppevent_del(reactor->epfd, ev);

    if (len > 0) {
        ev->length = len;
        ev->buffer[len] = '\0';
        // printf("C[%d]:%s\n", fd, ev->buffer);
        zeppevent_set(ev, fd, send_cb, reactor);
        zeppevent_add(reactor->epfd, EPOLLOUT, ev);
    } else if (len == 0) {
        close(ev->fd);
#if C100K
        printf("[fd=%d] closed\n", fd);
#else
        // printf("[fd=%d] pos[%ld], closed\n", fd, ev - reactor->events);
#endif
    } else {
        close(ev->fd);
        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
    }

    return (int)len;
}


int send_cb(int fd, uint32_t events, void *arg) {
    if (arg == NULL)
        return -1;
    struct zeppreactor *reactor = (struct zeppreactor *)arg;

#if C100K
    struct zeppevent *ev = zeppreactor_idx(reactor, fd);
#else
    struct zeppevent *ev = reactor->events + fd;
#endif

    ssize_t len = send(ev->fd, ev->buffer, ev->length, 0);
    if (len > 0) {
        // printf("send[fd=%d], [%ld]%s\n", fd, len, ev->buffer);

        zeppevent_del(reactor->epfd, ev);
        zeppevent_set(ev, fd, recv_cb, reactor);
        zeppevent_add(reactor->epfd, EPOLLIN, ev);
    } else {
        zeppevent_del(reactor->epfd, ev);
        close(ev->fd);
        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }

    return (int)len;
}


/// @brief 设置事件回调
/// @param ev 待设置事件
/// @param fd io
/// @param callback 回调
/// @param arg 回调参数
void zeppevent_set(struct zeppevent *ev, int fd, CALLBACK callback, void *arg) {
    ev->fd = fd;
    ev->callback = callback;
    ev->events = 0;
    ev->arg = arg;
    ev->last_active = time(NULL);
}


/// @brief 注册事件
/// @param epfd epoll句柄
/// @param events 注册事件的类型
/// @param ev 待注册的事件
/// @return 0 success -1 fail
int zeppevent_add(int epfd, uint32_t events, struct zeppevent *ev) {
    struct epoll_event epev = {0, {0}};

    epev.data.ptr = ev;
    epev.events = ev->events = events;

    //判断该事件是否已经添加过
    int op;
    if (ev->status & ZEPPEVENT_EXSIT) {
        op = EPOLL_CTL_MOD;
    } else {
        op = EPOLL_CTL_ADD;
        ev->status |= ZEPPEVENT_EXSIT;
    }

    if (epoll_ctl(epfd, op, ev->fd, &epev) < 0) {
        printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
        return -1;
    }
    return 0;
}


int zeppevent_del(int epfd, struct zeppevent *ev) {
    struct epoll_event epev = {0, {0}};
    if (ev->status & ZEPPEVENT_NEW) {
        return -1;
    }
    epev.data.ptr = ev;
    ev->status = ZEPPEVENT_NEW;
    epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &epev);
    return 0;
}


int zeppreactor_run(struct zeppreactor *reactor) {
    if (reactor == NULL)
        return -1;
    if (reactor->epfd < 0)
        return -1;
#if C100K
    if (reactor->evblk == NULL) {
        return -1;
    }
#else
    if (reactor->events == NULL)
        return -1;
#endif
    struct epoll_event events[MAX_EPOLL_EVENTS + 1];

    int checkpos = 0, i;
#if BENCHMARKING
    gettimeofday(&tv_begin, NULL);
#endif
    while (1) {
#if !BENCHMARKING
        long now = time(NULL);

        for (i = 0; i < 100; i++, checkpos ++) {
            if (checkpos == MAX_EPOLL_EVENTS) {
                checkpos = 0;
            }
            
            struct zeppevent *ev = zeppreactor_idx(reactor, checkpos);
            
            if (ev->status == 0 || ev->status & ZEPPEVENT_ISSOCKET) {
                continue;
            }

            long duration = now - ev->last_active;

            if (duration >= 60) {
                close(zeppreactor_idx(reactor, checkpos)->fd);
                printf("[fd=%d] timeout\n", ev->fd);
                zeppevent_del(reactor->epfd, ev);
            }
        }
#endif
        int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
        if (nready < 0) {
            printf("epoll_wait error, exit\n");
            continue;
        }

        for (int i = 0; i < nready; i++) {
            struct zeppevent *ev = (struct zeppevent *)events[i].data.ptr;

            // () && () --> ??
            // 第一个条件 (events[i].events & EPOLLIN) 检查当前事件是否为可读事件，
            // 而第二个条件 (ev->events * EPOLLIN) 确保当前注册的事件也是可读的。
            // 这种匹配确保了回调仅在相关事件发生时触发。
            if ((events[i].events & EPOLLIN) && (ev->events * EPOLLIN)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
            if ((events[i].events & EPOLLOUT) && (ev->events * EPOLLOUT)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }

        }
    }
}


int zeppreactor_addlistener(struct zeppreactor *reactor, int sockfd, CALLBACK callback) {
    if (reactor == NULL)
        return -1;
#if C100K
    if (reactor->evblk == NULL) {
        return -1;
    }
    if (reactor->blkcnt == 0) {
        return -1;
    }
    struct zeppevent *ev = zeppreactor_idx(reactor, sockfd);
    zeppevent_set(ev, sockfd, accept_cb, reactor);
    zeppevent_add(reactor->epfd, EPOLLIN, ev);
    ev->status |= ZEPPEVENT_ISSOCKET;
#else
    if (reactor->events == NULL)
        return -1;
    zeppevent_set(&reactor->events[sockfd], sockfd, accept_cb, reactor);
    zeppevent_add(reactor->epfd, EPOLLIN, &reactor->events[sockfd]);
#endif
    return 0;
}


int zeppreactor_init(struct zeppreactor *reactor) {
    if (reactor == NULL) {
        return -1;
    }

    memset(reactor, 0, sizeof(struct zeppreactor));
    reactor->epfd = epoll_create(1);

    if (reactor->epfd <= 0) {
        printf("[ERROR]create epfd in %s err %s\n", __func__, strerror(errno));
        return -2;
    }
#if C100K
    struct eventblock *blk = (struct eventblock *)malloc(sizeof(struct eventblock));
    if (blk == NULL) {
        printf("[ERROR]zeppreactor_init alloc block failed\n");
        return -2;
    }
    memset(blk, 0, sizeof(struct eventblock));

    struct zeppevent *evs = (struct zeppevent *)malloc(MAX_EPOLL_EVENTS * sizeof(struct zeppevent));
    if (evs == NULL) {
        printf("[ERROR]zeppreactor_init alloc block failed\n");
        return -2;
    }
    memset(evs, 0, MAX_EPOLL_EVENTS * sizeof(struct zeppevent));

    blk->events = evs;
    blk->next = NULL;

    reactor->evblk = blk;
    reactor->blkcnt = 1;
#else
    reactor->events = (struct zeppevent *)malloc(MAX_EPOLL_EVENTS * sizeof(struct zeppevent));
    if (reactor->events == NULL) {
        printf("[ERROR]zeppreactor_init alloc ractor.events faild\n")
    }
#endif
    return 1;
}


void zeppreactor_destory(struct zeppreactor *reactor) {
    close(reactor->epfd);
#if C100K
    struct eventblock *blk = reactor->evblk;
    struct eventblock *blknext = NULL;
    while (blk != NULL) {
        blknext = blk->next;
        free(blk->events);
        free(blk);
        blk = blknext;
    }
#else
    free(reactor->events);
#endif
}

#if C100K
int zeppreactor_alloc(struct zeppreactor *reactor) {
    if (reactor == NULL)
        return -1;
    if (reactor->evblk == NULL)
        return -1;

    struct eventblock *blk = reactor->evblk;
    while (blk->next != NULL) {
        blk = blk->next;
    }

    struct eventblock *block = (struct eventblock *)malloc(sizeof(struct eventblock));
    if (block == NULL) {
        printf("zeppreactor_alloc eventblock failed\n");
        return -2;
    }
    memset(block, 0, sizeof(struct eventblock));


    struct zeppevent *evs = (struct zeppevent *)malloc(MAX_EPOLL_EVENTS * sizeof(struct zeppevent));
    if (evs == NULL) {
        printf("zeppreactor_alloc zeppevents failed\n");
        return -2;
    }
    memset(evs, 0, (MAX_EPOLL_EVENTS * sizeof(struct zeppevent)));

    block->next = NULL;
    block->events = evs;


    blk->next = block;
    reactor->blkcnt ++;
    return 0;
}

struct zeppevent *zeppreactor_idx(struct zeppreactor *reactor, int sockfd) {
    int blkidx = sockfd / MAX_EPOLL_EVENTS;

    while (blkidx >= reactor->blkcnt) {
        zeppreactor_alloc(reactor);
    }

    int i = 0;
    struct eventblock *blk = reactor->evblk;
    while (i ++ < blkidx && blk != NULL) {
        blk = blk->next;
    }

    return &blk->events[sockfd % MAX_EPOLL_EVENTS];
}
#endif


int init_socket(int port) {
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);

    struct sockaddr_in serveraddr;
    memset(&serveraddr, 0, sizeof(struct sockaddr_in));

    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
    serveraddr.sin_port = htons(port);

    if (-1 == bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(struct sockaddr))) {
        perror("bind");
        return -1;
    }

    if (listen(sockfd, 20) < 0) {
        printf("listen failed : %s\n", strerror(errno));
    }

    return sockfd;
}

#if 0
int main() {

    struct zeppreactor *reactor = (struct zeppreactor *)malloc(sizeof(struct zeppreactor));
    zeppreactor_init(reactor);

    int sockfd = init_socket(DEFAULT_PORT);
    zeppreactor_addlistener(reactor, sockfd, accept_cb);

    zeppreactor_run(reactor);
    zeppreactor_destory(reactor);

    getchar();

    return 0;
}
#endif
